package com.tools;

import com.alibaba.fastjson.JSONObject;
import com.aliyun.datahub.client.exception.*;
import com.aliyun.datahub.client.model.BlobRecordData;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.clientlibrary.config.ConsumerConfig;
import com.aliyun.datahub.clientlibrary.config.ProducerConfig;
import com.aliyun.datahub.clientlibrary.consumer.Consumer;
import com.aliyun.datahub.clientlibrary.producer.Producer;
import com.bean.SignsDataBO;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;



/**
 * Author: kannabis
 * Date: 2024/1/25 22:25
 */
public class TestDatahubServiceV3 {

    public static final String accessId = "iFFSwUPZemOfRkfB";
    public static final String accessKey = "46aE5wnfkJjfFJQnY2cycAkhEnZAau";
    public static final String endpoint = "http://datahub.cn-cq-xczwy-d01.dh.res.alicloud.cqxczwy.com";
    public static final String projectName = "qx_wsx_dsjzx_datahub";
    public static final String topicName = "ods_wsx_zfb_cstz_wsqylydsjzhgkpt_zbz_delta";
    public static final String subId = "";


    public static final String appCode = "A50023700091760202309005205";
    public static final String version = "1.0";
    public static final String idxCode = "Q170410121";
    public static final String idxName = "超出人数";
//    public static final String idxDataID = idxCode+"_"+new Random().nextInt(100); //待填写
    public static final String pareZoningCode = "500000";
    public static final String zoningCode = "500237000000";
    public static final String zoningName = "巫山县";
    public static final String idxValue = String.valueOf(new Random().nextInt(100));

    public static SignsDataBO generateData(){
        String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        String idxDataID = idxCode+"_"+new Random().nextInt(100);
        SignsDataBO dataDTO = SignsDataBO.builder().
                version(version).
                appCode(appCode).
                idxCode(idxCode).
                idxName(idxName).
                idxDataId(idxDataID).
                parentZoningCode(pareZoningCode).
                zoningCode(zoningCode).
                zoningName(zoningName).

                idxValue("1832").
                idxTime(time).
                valueResult("1").
                circleComparePositive("+").
                sameComparePositive("-").
                build();
        return dataDTO;
    }

    public static Producer initProducer() {
        //初始化生产者
        ProducerConfig pConfig = new ProducerConfig(endpoint, accessId, accessKey);
        return new Producer(projectName, topicName, pConfig);
    }

    public static Consumer initConsumer() {
        //初始化消费者
        ConsumerConfig cConfig = new ConsumerConfig(endpoint, accessId, accessKey);
        return new Consumer(projectName, topicName, subId, cConfig);
    }

    public static List<String> generateDataList(){
        List<String > list = new ArrayList<>();
        SignsDataBO signsDataBO1 = generateData();
        String jsonString1 = JSONObject.toJSONString(signsDataBO1);
        System.out.println(jsonString1);
        SignsDataBO signsDataBO2 = generateData();
        String jsonString2 = JSONObject.toJSONString(signsDataBO2);
        System.out.println(jsonString2);
        SignsDataBO signsDataBO3 = generateData();
        String jsonString3 = JSONObject.toJSONString(signsDataBO3);
        System.out.println(jsonString3);
        list.add(jsonString1);
        list.add(jsonString2);
        list.add(jsonString3);
        return list;
    }

    //发送Blob消息
    public static void sendBlob(List<String> dataList) {
        if(dataList.isEmpty()){
            return;
        }
        List<RecordEntry> list = new ArrayList<>();
        for (String str : dataList) {
            if (StringUtils.isNotBlank(str)) {
                try {
                    RecordEntry recordEntry = new RecordEntry();
                    BlobRecordData blobRecordData = new BlobRecordData
                            (str.getBytes(StandardCharsets.UTF_8));
                    recordEntry.setRecordData(blobRecordData);
                    list.add(recordEntry);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        Producer producer = initProducer();
        producer.send(list, 10);
    }

    //消费
    public void toConsumer() {
        System.out.println("===============开始读取数据");
        Consumer consumer = initConsumer();
        int maxRetry = 1000;
        boolean stop = false;
        try {
            while (!stop) {
                try {
                    while (true) {
// 协同消费刚初始化，需要等待服务端分配shard，约40秒，期间只能返回null
// 自动提交模式，每次调用read，认为之前读的数据都已处理完成，自动ack
                        RecordEntry record = consumer.read(maxRetry);
// 处理数据
                        if (record != null) {
                            BlobRecordData data = (BlobRecordData) record.getRecordData();
// 根据自己的schema来处理数据，此处打印第一列的内容
                            System.out.println(StringUtils.toEncodedString(data.getData(), Charset.defaultCharset()));
// 根据列名取数据
// System.out.println("field2: {}", data.getField("field2"));
// 非自动提交模式，每条record处理完后都需要ack
// 自动提交模式，ack不会做任何操作
// 1.1.7版本及以上
                            record.getKey().ack();
                        } else {
                            System.out.println("read null");
                        }
                    }
                } catch (SubscriptionOffsetResetException e) {
// 点位被重置，重新初始化consumer
                    try {

                    } catch (DatahubClientException e1) {
// 初始化失败，重试或直接抛异常
                        throw e;
                    }
                } catch (InvalidParameterException |
                        SubscriptionOfflineException |
                        SubscriptionSessionInvalidException |
                        AuthorizationFailureException |
                        NoPermissionException e) {
// 请求参数非法
// 订阅被下线
// 订阅下相同shard被其他客户端占用
// 签名不正确
// 没有权限
                    throw e;
                } catch (DatahubClientException e) {
// 基类异常，包含网络问题等，可以选择重试
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
// 确保资源正确释放
// 会提交已ack的点位
            consumer.close();
        }

    }
}
